package cn.doit.day05

import scala.collection.immutable
import scala.io.Source

//1.求每部电影的平均分，按照降序输出  ==> 我需要得是电影名称 + 平均分
/*
    1.读取数据
    2.对数据进行处理  etl  ==》 元组
    3.对于评分这个数据集操作==》 平均分   分数得总和 / 评分得次数
    4.对于评分这张表进行groupBy ==》 电影id ==》 Map[String(电影id),List[(arr(0), arr(1), arr(2), arr(3))]]
    5.用map算子对于value得值(list集合)进行转变 ==》 List[Double]  电影id，list.sum/list.size
    6.将电影id转变成电影名称  ==》 map去get数据 ==》 电影得id是唯一得
 */




object 电影案例 {
  def main(args: Array[String]): Unit = {
    //求电影得平均分  总分数/评分次数  ==》 将电影得所有评分都放在一个list集合中
    val ratings: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    val movies: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val users: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\users.txt").getLines().toList

    //对数据集进行处理
    val ratings_etl: List[(String, String, String, String)] = ratings.map(line => {
      val arr: Array[String] = line.split("_")
      (arr(0), arr(1), arr(2), arr(3))
    })

    val movies_etl: Map[String, (String, String)] = movies.map(line => {
      val arr: Array[String] = line.split("_")
      (arr(0), (arr(1), arr(2)))
    }).toMap

    val users_etl: Map[String, (String, String, String, String)] = users.map(line => {
      val arr: Array[String] = line.split("_")
      (arr(0), (arr(1), arr(2), arr(3), arr(4)))
    }).toMap

    //需求1：求每部电影的平均分，按照降序输出  ==> 我需要得是电影名称 + 平均分
    /*   val list: List[(String, Double)] = getMovieAvgScore(ratings_etl, movies_etl)
      .toList

    list
      .sortBy(-_._2)
      .take(10)
      .foreach(println)

    println("==============")

    list
      .sortBy(_._2)
      .take(10)
      .foreach(println)*/

    //2.求每个用户评过电影的总数，按照降序排列
    /*
    userRatingCount(ratings_etl)
      .toList
      .sortBy(_._2)
      .reverse
      .take(10)
      .foreach(println)
*/
    //3.每个用户最早评价的前三部电影(用户，电影信息，评分值)
   /* top3Movie(ratings_etl, movies_etl)
      .toList
      .sortBy(_._1)
      .take(10)
      .foreach(println)*/

    //4.哪些年份的电影评分(平均分)最高，取最高的前五个年份
   /* yearAvgScore(ratings_etl, movies_etl)
      .foreach(println)*/

    //5.每个电影类型的平均分按照降序排列取前3
    /*    top3MovieType(ratings_etl, movies_etl)
          .toList
          .sortBy(_._2)
          .reverse
          .take(3)
          .foreach(println)*/

    //6.每个职业最喜欢的前三个电影类型(按照观看次数排)
    /*    top3MostLoveMovieType(ratings_etl, movies_etl,users_etl)
      .foreach(println)*/

    //7.年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影（Id）
    /*    mMostMovie(ratings_etl,users_etl)
          .foreach(println)*/



  }

  def getMovieAvgScore(ratings_etl: List[(String, String, String, String)], movies_etl: Map[String, (String, String)]) = {
    //根据movie_id 进行分组
      ratings_etl.groupBy(_._2)//Map[电影id，List(四元元组)]
        .map(tp => {
          //获取到movie_id
          val movie_id: String = tp._1
          //所有分数得汇总
          val ratings: List[Double] = tp._2.map(_._3.toDouble)
          //将电影id转换成电影名称
          val movieNameAndType: (String, String) = movies_etl.getOrElse(movie_id, (null, null))
          //电影名称
          (movieNameAndType._1, (ratings.sum/ratings.size).formatted("%.2f").toDouble)
        })
  }

  //2.求每个用户评过电影的总数，按照降序排列
  def userRatingCount(ratings_etl: List[(String, String, String, String)])={
    val userAndRatings: Map[String, List[(String, String, String, String)]] = ratings_etl.groupBy(_._1)
    //对value得值进行转变你，里面只放评论的电影id
    userAndRatings.map(tp=>(tp._1,tp._2.map(_._2).distinct.size))
  }

  //3.每个用户最早评价的前三部电影(用户，电影信息，评分值)
  def top3Movie(ratings_etl: List[(String, String, String, String)], movies_etl: Map[String, (String, String)])={
    //按照用户进行分组
    val userAndRatings: Map[String, List[(String, String, String, String)]] = ratings_etl.groupBy(_._1)
    //需要让他按照时间戳的升序进行排列，排完了之后取前三
    userAndRatings.map(tp=>{
      val user_id: String = tp._1
      //先按照时间戳的升序进行排列，然后取前三
      val ratings_info: List[(String,String, String, String)] = tp._2.sortBy(_._4).take(3)
      val tuples: List[(String, String, String)] = ratings_info.map(tp => {
        val movie_id: String = tp._2
        val rating: String = tp._3
        val movieNameAndType: (String, String) = movies_etl.getOrElse(movie_id, (null, null))
        (movieNameAndType._1, movieNameAndType._2, rating)
      })
      (user_id,tuples)
    })
  }

  //4.哪些年份的电影评分(平均分)最高，取最高的前五个年份
  //将所有的电影id全部换成年份
  //要将movies 这个数据源中的年份给提取出来  Toy Story (1995)  ==》 1995
  def yearAvgScore(ratings_etl: List[(String, String, String, String)], movies_etl: Map[String, (String, String)])={
    //把年份取出来
    val idAndYear: Map[String, String] = movies_etl.map(tp => {
      val movie_id: String = tp._1
      val movie_name: String = tp._2._1
      val year: String = movie_name.substring(movie_name.length - 5, movie_name.length - 1)
      (movie_id, year)
    })

    val yearAndScore: List[(String, String)] = ratings_etl.map(tp => {
      val movie_id: String = tp._2
      val year: String = idAndYear.getOrElse(movie_id, "9999")
      (year, tp._3)
    })

    yearAndScore.groupBy(_._1)
      .map(tp=>(tp._1,(tp._2.map(_._2.toDouble).sum/tp._2.size).formatted("%.2f").toDouble))
      .toList
      .sortBy(_._2)
      .reverse
      .take(5)
  }

    //5.每个电影类型的   平均分按照      降序排列   取前3
    //电影类型属于movie.txt   arr(2)=>电影类型   电影id   1   ==》    5     四个电影类型都是5分  一对多的关系
    def top3MovieType1(ratings_etl: List[(String, String, String, String)], movies_etl: Map[String, (String, String)])={
      //第二步：从另外一个数据源ratings ==》 获取电影id和分数   List((1,5),(2,3)....) ==> List((5,Animation) (3,Children's) (4,Comedy).....)
      //对应下面的groupBy
      val idAndRatings: Map[String, List[String]] = ratings_etl.groupBy(_._2) //==>Map[电影id，List[四元元祖]]
        .map(tp => {
          val movie_id: String = tp._1
          val ratings: List[String] = tp._2.map(_._3)
          (movie_id, ratings)
        })

      //第一步：对于数据源针对该需求做一个数据的etl处理 ==》 List((1,Animation) (1,Children's) (1,Comedy))
      val tuples: List[(String, List[String])] = movies_etl.toList.flatMap(tp => {
        val movie_id: String = tp._1
        val ratings: List[String] = idAndRatings.getOrElse(movie_id, null)
        val nameAndType: (String, String) = tp._2
        val typeArr: Array[String] = nameAndType._2.split("\\|")
        //为什么能想到这个for循环 ==》   List((1,Animation) (1,Children's) (1,Comedy))
        for (movieType <- typeArr) yield {
          (movieType, ratings)
        }
      })

      //对于数据源最终的处理
      val typeAndRating: List[(String, Double)] = tuples.flatMap(tp => {
        for (elem <- tp._2) yield {
          (tp._1, elem.toDouble)
        }
      })



    }



    //(1 ,(Toy Story (1995)    Animation|Children's|Comedy))  ==>  (1,Animation|Children's|Comedy) ==>(1,Animation) (1,Children's) (1,Comedy)
    //目标:电影类型的平均分  电影类型   分数   算平均分....





  //5.每个电影类型的平均分按照降序排列取前3
  def top3MovieType(ratings_etl: List[(String, String, String, String)], movies_etl: Map[String, (String, String)])={
    //将电影id和电影类型拼接后返回
    val idAndType: Map[String, Array[String]] = movies_etl.map(tp => {
      //将电影类型切开
      val typeArr: Array[String] = tp._2._2.split("\\|")
      (tp._1, typeArr)
    })

    val typeAndRating: List[(String, String)] = ratings_etl.flatMap(tp => {
      val movie_type: Array[String] = idAndType.getOrElse(tp._2, Array("未知"))
      for (elem <- movie_type) yield {
        (elem, tp._3)
      }
    })
    typeAndRating.groupBy(_._1)
      .map(tp=>(tp._1,(tp._2.map(_._2.toDouble).sum/tp._2.size).formatted("%.2f").toDouble))
  }

  //6.每个职业最喜欢的前三个电影类型(按照观看次数排)
  //1.对数据进行etl清洗  ==》 我需要将users表中的user_id和job
  //2.对于ratings这张表      user_id  和  movie_id
  //3.对于movies这张表       movie_id 和  movie_type
  //==> 以ratings作为主表  user_id  和  movie_id    getorEles()==>job  电影类型 ==》(job，电影类型=》需求切割 flatmap)





  def top3MostLoveMovieType(ratings_etl: List[(String, String, String, String)], movies_etl: Map[String, (String, String)],users_etl: Map[String, (String, String, String, String)])={
    //将电影id和电影类型拼接后返回
    val idAndType: Map[String, Array[String]] = movies_etl.map(tp => {
      //将电影类型切开
      val typeArr: Array[String] = tp._2._2.split("\\|")
      (tp._1, typeArr)
    })

    //再将电影id转换成电影类型
    val typeAndRating: List[(String, String)] = ratings_etl.flatMap(tp => {
      //将user_id转换成职业
      val user_id: String = tp._1
      //直接获取到职业
      val job: String = users_etl.getOrElse(user_id, (null, null, null, null))._3
      val movie_type: Array[String] = idAndType.getOrElse(tp._2, Array("未知"))

      for (movie_type <- movie_type) yield {
        (movie_type, job)
      }
    })

    typeAndRating.groupBy(_._2)
      .map(tp=>{
        val job: String = tp._1
        val listMovieType: List[String] = tp._2.map(_._1)
        val top3: List[(String, Int)] = listMovieType.groupBy(word => word)
          .map(tp => (tp._1, tp._2.size))
          .toList
          .sortBy(_._2)
          .reverse
          .take(3)
        (job,top3)
      })

  }


  //7.年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影（Id）
  //user表中作一个过滤==》过滤出18-24的年轻人，还要过滤男性  ==》filter  ==》 Map(user_id，null)
  //rating表中 user_id，movie_id  ==> 怎么在ratings这张表里面过滤出来18-24的男性这群人
  def mMostMovie(ratings_etl: List[(String, String, String, String)],users_etl: Map[String, (String, String, String, String)])={
    //过滤出来所有符合条件的男性
    val m_18: Map[String, (String, String, String, String)] = users_etl.filter(tp => {
      val value: (String, String, String, String) = tp._2
      //
      value._1.equals("F") && value._3.equals("18")
    })

    //过滤出来所有的符合年龄段的男性用户所看过的电影
    val movieFiltered: List[(String, String, String, String)] = ratings_etl.filter(tp => {
      //从符合条件的男性集合中获取数据
      val maybeTuple: Option[(String, String, String, String)] = m_18.get(tp._1)
      //判断下能否获取到数据，如果获取到数据代表这个user_id符合条件，否则不符合，不符合的就过滤掉
      maybeTuple.isDefined
    })

    //根据电影id进行分组，对用户进行计算排序
    movieFiltered.groupBy(_._2)
      .map(tp=>{
        val movie_id: String = tp._1
        val users: Set[String] = tp._2.map(_._1).toSet
        (movie_id,users.size)
      })
      .toList
      .sortBy(-_._2)
      .take(10)
  }

}
